package com.lagou.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import static java.lang.System.out;

public class HdfsClientDemo {
    FileSystem fileSystem = null;
    Configuration conf = null;
    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        // 1 获取Hadoop 集群的Configuration对象
        conf = new Configuration();
        //conf.set("fs.defaultFS","hdfs://centos7-1:9000");
        conf.set("dfs.replication","3");
        // 2 根据Configuration获取FileSystem对象
        fileSystem = FileSystem.get(new URI("hdfs://centos7-1:9000"), conf, "root");

    }

    @Test
    public void testMkdis() throws URISyntaxException, IOException, InterruptedException {
        // 3 使用FileSystem对象创建一个测试目录
        fileSystem.mkdirs(new Path("/api_test"));
    }

    @Test
    public void testCopyFromLocalToHdfs() throws URISyntaxException, IOException, InterruptedException {
        // 上传文件
        // src:源文件目录:本地路径
        // dst:目标文件目录:hdfs目录
        fileSystem.copyFromLocalFile(new Path("D:/hadoop.txt"), new Path("/api_test/lagou.txt"));
    }

    @Test
    public void testCopyFromHdfsToLocal() throws URISyntaxException, IOException, InterruptedException {
        // 下载文件
        // boolean:是否删除源文件
        // src:源文件目录:hdfs目录
        // dst:目标文件目录:本地路径
        fileSystem.copyToLocalFile(true, new Path("/lagou.txt"), new Path("D:/hadoop.txt"));
    }

    @Test
    public void testDeleteFile() throws URISyntaxException, IOException, InterruptedException {
        // 删除hdfs上的文件或文件夹
        // dst:hdfs目标文件或目录路径
        // boolean:是否递归删除
        fileSystem.delete(new Path("/lagou.txt"), true);
    }

    @Test
    public void testListFiles() throws URISyntaxException, IOException, InterruptedException {
        // 遍历hdfs的根目录得到文件及文件夹的信息: 名称、权限、长度等
        // dst:hdfs目标文件或目录路径
        // boolean:是否递归删除
        RemoteIterator<LocatedFileStatus> filesIterator = fileSystem.listFiles(new Path("/lagou"), true);
        while(filesIterator.hasNext()){
            LocatedFileStatus fileStatus = filesIterator.next();
            // 文件名
            String fileName = fileStatus.getPath().getName();
            
            // 权限
            FsPermission permission = fileStatus.getPermission();
            
            // 分组
            String group = fileStatus.getGroup();

            // 用户
            String owner = fileStatus.getOwner();
            
            // 块信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    out.println("主机名" + host);
                }
            }
        }
    }

    @Test
    public void testIsFile() throws URISyntaxException, IOException, InterruptedException {
        // hdfs上的文件或文件夹的判断
        // dst:hdfs目标文件或目录路径
        // 占时没有现成的递推判断
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/lagou.txt"));
        for (FileStatus fileStatus : fileStatuses) {
            // 判断是否是文件
            boolean flag = fileStatus.isFile();
            if(flag){
                // 文件
                out.println("文件：" + fileStatus.getPath().getName());
            }else {
                // 文件夹
                out.println("文件夹：" + fileStatus.getPath().getName());

            }
        }
    }


    @Test
    public void testUpLoadFileIO() throws URISyntaxException, IOException, InterruptedException {
        // 使用IO流操作HDFS
        // 上传文件：准备输入流读取本地文件，使用Hdfs的输出流写数据到hdfs
        // 1 读取本地文件的输入流
        FileInputStream fis = new FileInputStream(new File("D:/a.txt"));
        // 2 准备写数据到hdfs的输出流
        FSDataOutputStream fsOutputStream = fileSystem.create(new Path("/api_test/lagou_a.txt"));

        // 3 输入流数据拷贝到输出流，默认会自动关闭流
        IOUtils.copyBytes(fis, fsOutputStream, conf);
    }

    @Test
    public void testDownLoadFileIO() throws URISyntaxException, IOException, InterruptedException {
        // 使用IO流操作HDFS
        // 1 读取hdfs文件的输入流
        FSDataInputStream in = fileSystem.open(new Path("/api_test/lagou.txt"));

        // 2 准备本地文件的输出流
        FileOutputStream out = new FileOutputStream(new File("D:/lagou.txt"));

        // 3 输入流数据拷贝到输出流，默认会自动关闭流
        IOUtils.copyBytes(in, out, conf);
    }


    @Test
    public void testUploadPacket() throws URISyntaxException, IOException, InterruptedException {
        // 演示从本地文件系统上传文件到dfs的过程
        // 1 读取本地文件的输入流
        FileInputStream in = new FileInputStream(new File("D:/lagou.txt"));

        // 2 准备写出数据到hdfs的输出流
        FSDataOutputStream out = fileSystem.create(new Path("/api_test/lagou_test1.txt"), new Progressable() {
            // 这个progress方法就是每传输64KB（一个packet）就会执行一次
            public void progress() {
                System.out.println("&&");
            }
        });

        // 3 输入流数据拷贝到输出流，默认会自动关闭流
        IOUtils.copyBytes(in, out, conf);
    }

    @After
    public void destory() throws IOException {
        // 4 释放FileSystem对象
        fileSystem.close();
    }
}
